package com.swak.telnet.invoker;

import java.util.concurrent.CompletableFuture;

import com.swak.Constants;
import com.swak.exception.RemotingException;
import com.swak.exception.RpcException;
import com.swak.exception.TimeoutException;
import com.swak.loadbalance.LoadBalance;
import com.swak.loadbalance.LoadBalances;
import com.swak.reactivex.transport.ExchangeClient;
import com.swak.registry.URL;
import com.swak.telnet.cmd.Command;
import com.swak.utils.Lists;

/**
 * 发送Telnet命令
 * 
 * @author lifeng
 */
public class TelnetInvoker {

	private final ExchangeClient[] clients;
	private final LoadBalance<ExchangeClient> loadBalance;
	private final URL url;

	public TelnetInvoker(URL url, ExchangeClient[] clients) {
		this.url = url;
		this.clients = clients;
		this.loadBalance = LoadBalances.RandomLoad.newLoadBalance();
		this.loadBalance.onRefresh(Lists.newArrayList(clients));
	}

	/**
	 * 发送请求
	 * 
	 * @param command 命令行
	 * @return
	 */
	public CompletableFuture<Object> invoke(Command command) throws RpcException {
		ExchangeClient client = loadBalance.select();
		boolean twoWay = url.getParameter(Constants.TwoWay, Boolean.TRUE);
		int timeout = url.getParameter(Constants.Timeout, Constants.Default_Timeout);
		try {
			if (twoWay) {
				return client.request(command, timeout);
			}
			client.send(command, false);
			return CompletableFuture.completedFuture(null);
		} catch (TimeoutException e) {
			throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke command timeout. command: "
					+ command.toString() + ", provider: " + url + ", cause: " + e.getMessage(), e);
		} catch (RemotingException e) {
			throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote command: "
					+ command.toString() + ", provider: " + url + ", cause: " + e.getMessage(), e);
		}
	}

	public URL url() {
		return url;
	}

	public void destroy() {
		try {
			for (ExchangeClient client : clients) {
				client.close();
			}
		} catch (Exception e) {
		}
	}
}